package flink.windowCount;

import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Date;

/**
 * 更据作业运行日志,
 * 统计各个时间段作业执行多少作业,
 * 需要使用eventTime
 * 需要使用state
 */
public class windowCountApp {

    private static Integer cnt = 0;

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

//        DataStreamSource<String> source = env.readTextFile("D:\\Destop\\catalina.out");
        DataStreamSource<String> source = env.readTextFile("D:\\Destop\\20210630.txt");
        source.filter(new filters())
                .map(new sourceMap())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Job>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                                .withTimestampAssigner((job, timestamp) -> job.getLogTime())
                )
                .windowAll(TumblingEventTimeWindows.of(Time.seconds(300)))
                .apply(new AllWindowFunction<Job, Tuple2<String, Integer>, TimeWindow>() {
                    @Override
                    public void apply(TimeWindow window, Iterable<Job> elements, Collector<Tuple2<String, Integer>> out) throws Exception {

                        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
                        String start = simpleDateFormat.format(window.getStart());
                        String end = simpleDateFormat.format(window.getEnd());
                        if (start.endsWith("00:00:00.000")) {
                            cnt = 1;

                        }

                        int acc = cnt;
                        for (Job element : elements) {
//                            System.out.println(element.toString());

                            if (element.getType() > 0) {
                                acc++;
                            }
                            cnt = cnt + element.getType();
                        }

                        System.out.println(String.format("==================> 当前窗口的开始时间: %s,结束时间:%s,已执行任务数:%s,正在执行的任务数:%s", start, end, acc, cnt));
                        out.collect(new Tuple2<>(end, acc));
                    }
                })
                .print("==================");

        env.execute();


    }

    private static class sourceMap implements MapFunction<String, Job> {
        @Override
        public Job map(String s) throws Exception {
            SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
            String[] split = s.split(" ");
            if (split.length == 16) {
                String strTime = split[0] + " " + split[1];
                Date parse = simpleDateFormat.parse(strTime);
                Long jobId = Long.parseLong(split[14].replace(",address", ""));
                return new Job(jobId, strTime, parse.getTime(), 1);
            } else {
                String strTime = split[0] + " " + split[1];
                Date parse = simpleDateFormat.parse(strTime);
                Long jobId = Long.parseLong(split[6].substring(4, 22));
                return new Job(jobId, strTime, parse.getTime(), -1);
            }

        }
    }

    private static class filters implements FilterFunction<String> {
        @Override
        public boolean filter(String s) throws Exception {
            if (s.contains("master send debug command to worker")) {
                return true;
            } else if (s.contains("执行结果")) {
                return true;
            }
            return false;
        }
    }


    public static class Job {
        private Long jobId;
        private String strTime;
        private long logTime;
        private Integer type;

        public Job() {
        }

        public Job(Long jobId, String strTime, long logTime, Integer type) {
            this.jobId = jobId;
            this.strTime = strTime;
            this.logTime = logTime;
            this.type = type;
        }

        public Long getJobId() {
            return jobId;
        }

        public void setJobId(Long jobId) {
            this.jobId = jobId;
        }

        public long getLogTime() {
            return logTime;
        }

        public void setLogTime(long logTime) {
            this.logTime = logTime;
        }

        public Integer getType() {
            return type;
        }

        public void setType(Integer type) {
            this.type = type;
        }

        public String getStrTime() {
            return strTime;
        }

        public void setStrTime(String strTime) {
            this.strTime = strTime;
        }

        @Override
        public String toString() {
            return JSON.toJSONString(this);
        }
    }
}
